package com.tyc.kafka.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * 发送-->拦截器-->序列化器-->分区器
 * 测试自定义分区器
 */
public class CustomProducerMyPartition {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //属性配置
        Properties properties = new Properties();
        // 连接地址 多个用逗号隔开
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092,127.0.0.1:9093");
        // 指定序列化类型
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 指定分区器
        properties.setProperty(ProducerConfig.PARTITIONER_CLASS_CONFIG,MyPartitioner.class.getName());
        //创建生产者
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        //同步发送数据
        for (int i = 0; i < 5; i++) {
            kafkaProducer.send(new ProducerRecord<>("test0", i+"hello"), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if(null == e){
                        System.out.println(String.format("主题：%s,分区：%s",recordMetadata.topic(),recordMetadata.partition()));
                    }

                }
            }).get();
        }
        //关闭资源
        kafkaProducer.close();
    }
}
